/**
 * 
 */
package org.sensmon.core.datamodel.filters;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @author LAP
 *
 */
public abstract class RunnableFilter<T> implements Runnable {
	
	private List<RunnableFilter<T>> nextFilters;
	private LinkedBlockingQueue<T> values;
	private boolean canceled;
	
	public RunnableFilter() {
		nextFilters = new LinkedList<RunnableFilter<T>>();
		values = new LinkedBlockingQueue<T>();
		canceled = false;
	}
	
	@Override
	public void run() {
		while(!canceled) {
			T value = tryTakeValue();
			value = doFilter(value);
			fireValue(value);
		}
	}
	
	private void fireValue(T value) {
		for(RunnableFilter<T> filter : nextFilters) {
			filter.addValue(value);
		}
	}

	protected abstract T doFilter(T data);
	
	public void addValue(T value) {
		tryPutValue(value);
	}
	
	private void tryPutValue(T value) {
		try {
			values.put(value);
		} catch (InterruptedException e) {
			// TODO: handle exception
		}
	}
	
	private T tryTakeValue() {
		T value = null;
		try {
			value = values.take();
		} catch (InterruptedException e) {
			// TODO: handle exception
		}
		return value;
	}
	
	public void addFilter(RunnableFilter<T> value) {
		nextFilters.add(value);
	}
	
	public void removeFilter(RunnableFilter<T> value) {
		nextFilters.remove(value);
	}
	
	public void removeFilter(int index) {
		nextFilters.remove(index);
	}
	
	public RunnableFilter<T> getRunnableFilter(int index) {
		return nextFilters.get(index);
	}
	
	public synchronized boolean isCanceled() {
		return canceled;
	}
	
	public synchronized void cancel() {
		canceled = true;
	}
}
